package ex.datastream;

import ex.vo.SensorReading;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public abstract class ApiFrame {
    protected StreamExecutionEnvironment env;

    protected void getEnv() {
        env = StreamExecutionEnvironment.getExecutionEnvironment();
    }

    protected KafkaSource getKafkaSource() {
        // 加载数据源
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("192.168.1.4:9092")
                .setTopics("test9")
                .setGroupId("my-group")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        return source;
    }

    protected  DataStream<Tuple2<String, Long>>  getsocketTextStreamText() {
        DataStream<String> lines = env.socketTextStream("localhost", 9999);
        DataStream<Tuple2<String, Long>> dataStream= lines.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Long>> out) throws Exception {
                String str[] = s.split(",");
                out.collect(Tuple2.of(str[0], Long.parseLong(str[1])));
            }
        });
        return dataStream;
    }
    protected  SingleOutputStreamOperator<Tuple2<String,Long>> getSingleOutputStreamOperator() {
        DataStream<String> lines = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<Tuple2<String, Long>> outputStreamOperator = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Long>> out) throws Exception {
                String str[] = s.split(" ");
                out.collect(Tuple2.of(str[0], Long.parseLong(str[1])));
            }
        });
        return outputStreamOperator;
    }


    protected  DataStream<SensorReading> getSensorReading() {
        DataStream<SensorReading> lines = env.socketTextStream("localhost", 9999).map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String s) throws Exception {
                String str[]=s.split(",");
                return new SensorReading(str[0],str[1],Double.valueOf(str[2]));
            }
        });
        return lines;
    }
}
